from airflow.providers.dingding.operators.dingding import DingdingHook
from airflow.providers.presto.hooks.presto import PrestoHook

from utils.macros import cst_ds_nodash


def es_threeCodeOnSuccess(kwargs, at_mobiles=None, at_all=None):
    def _es_threeCodeOnSuccess(ctx):
        # envInfo 环境基本信息
        env = ctx['var']['value'].env
        execution_date = cst_ds_nodash(ctx["execution_date"])

        # 监控表， 参数传参，库名，表名，钉钉机器人
        db = kwargs["db"]
        table = kwargs["table"]
        desc = kwargs["desc"]
        title = f'[{env.upper()}]: es二段码更新失败网点'
        dingding_conn_id = kwargs["dingding_conn_id"]

        # 使用presto计算 监控表日数据量
        try:
            presql = f"select count(1) from {db}.{table} where dt = '{execution_date}'"
            presql1 = f"select * from {db}.{table} where dt = '{execution_date}'"
            preconn = PrestoHook(presto_conn_id='presto_default')
            hivecnt = preconn.get_records(presql)[0][0]
            result = preconn.get_records(presql1)
            print(result)
            print(hivecnt)
        except Exception as err:
            print(err)

        if hivecnt > 0:
            dingding_conn_id = f'{dingding_conn_id}'
        else:
            dingding_conn_id = ""

        # 发送钉钉的任务日志
        text = f"""
# {title}
   - *表全名:* {db}.{table}
   - *分区:* {execution_date}
   - *分区数据量:* {hivecnt}
   - *详情:* {result}
   - *表描述:* {desc}
"""

        # dingding_conn_id = "dingding_last"
        return DingdingHook(dingding_conn_id=dingding_conn_id,
                            message_type='markdown',  # text, link, markdown, actionCard, feedCard
                            message={'title': title, 'text': text},
                            at_mobiles=at_mobiles,
                            at_all=at_all, ).send()

    return _es_threeCodeOnSuccess
